/**
 * 
 */
package io.gitee.falllee.dao.core.table;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.calcite.linq4j.Linq4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import io.gitee.falllee.dao.core.HeaderBuilder;
import io.gitee.falllee.dao.model.DaoConst;
import io.gitee.falllee.dao.model.DaoException;
import io.gitee.falllee.dao.persistence.DataMapper;
import io.gitee.falllee.dao.util.CommonUtils;
import io.gitee.falllee.dao.util.FilterPlus;

/**
 * @author batty
 * @date 2022-06-27
 *
 */
public class DeleteBuilder extends HeaderBuilder {

    private TableBuilder tableBuilder;
    
    private DataMapper dataMapper;
    
    private CacheBuilder cacheBuilder;
    
    private MessageBuilder messageBuilder;

    private String tableName;
    
    private String keyField;

    Boolean hardDelete = false;

    DeleteBuilder(TableBuilder parent) {
        super(parent.getDaoConfig());
        this.tableBuilder = parent;
        this.dataMapper = parent.getDataMapper();
        this.cacheBuilder = parent.cacheBuilder();
        this.messageBuilder = parent.messageBuilder();
        this.tableName = parent.getTableName();
        this.keyField = parent.getKeyField();
    }
    
    public DeleteBuilder header(String key, String value) {
        if (headerParamsMap == null) {
            headerParamsMap = new HashMap<>(16);
        }
        headerParamsMap.put(key, value);
        return this;
    }
    
    public DeleteFilterBuilder filter(FilterPlus filter) {
        return new DeleteFilterBuilder(this, daoConfig, filter);
    }

    /**
     * 是否硬删除，默认false
     * 
     * @return
     */
    public DeleteBuilder hardDelete() {
        this.hardDelete = true;
        return this;
    }

    public Boolean deleteByKey(Object keyValue) {
        if (keyValue == null) {
            throw new DaoException("deleteByKey keyValue can not be empty");
        }
        Boolean bl = deleteByKeyValue(keyValue);
        if (Boolean.TRUE.equals(bl)) {
            // 如果启用缓存，删除缓存数据
            cacheBuilder.clearCache(keyValue);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValue, DaoConst.DELETE);
        }
        return bl;
    }

    private Boolean deleteByKeyValue(Object keyValue) {
        // 默认软删除
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        map.put(DaoConst.FILTER, getParamExpression(keyField));
        map.put(keyField, keyValue);
        Integer res = 0;
        if (Boolean.TRUE.equals(hardDelete)) {
            res = dataMapper.delete(map);
        } else {
            setEnableFlag(map, false);
            setUpdateInfo(map);
            res = dataMapper.deleteLogic(map);
        }
        return res >= 0;
    }

    public Boolean deleteByKeyList(List<? extends Object> keyValueList) {
        List<? extends Object> tempKeyValueList = Linq4j.asEnumerable(keyValueList).where(p -> !ObjectUtils.isEmpty(p))
                .distinct().toList();
        if (CollectionUtils.isEmpty(tempKeyValueList)) {
            throw new DaoException("keyValueList can not be empty");
        }
        // 默认软删除
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        int max = DaoConst.SELECT_MAX_SIZE.intValue();
        int total = tempKeyValueList.size();
        int num = total / max;
        int left = total % max;
        num = left > 0 ? num + 1 : num;
        Integer res = 0;
        for (int i = 0; i < num; i++) {
            List<? extends Object> subKeyList = CommonUtils.getSubList(tempKeyValueList, max, i);
            String filter = getKeyListFilter(subKeyList, keyField, map);
            map.put(DaoConst.FILTER, filter);
            if (Boolean.TRUE.equals(hardDelete)) {
                res += dataMapper.delete(map);
            } else {
                setEnableFlag(map, false);
                setUpdateInfo(map);
                res += dataMapper.deleteLogic(map);
            }
        }
        if (res > 0) {
            // 如果启用缓存，删除缓存数据
            cacheBuilder.clearCache(tempKeyValueList);
            // 发送Kafka消息
            messageBuilder.sendMessage(tempKeyValueList, DaoConst.DELETE_LIST);
        }
        return res >= 0;
    }

    Boolean delete(FilterPlus filterPlus) {
        String filter = filterPlus.getParamFilter();
        if (!StringUtils.hasLength(filter)) {
            throw new DaoException("delete filter can not be empty");
        }
        // 默认软删除
        Map<String, Object> map = filterPlus.getParamMap();
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        map.put(DaoConst.FILTER, filter);
        // 逻辑删除需要设置enbaled_flag和update信息
        if (Boolean.TRUE.equals(hardDelete)) {
            delEnableFalg(map);
        } else {
            setEnableFlag(map, false);
        }
        // 先查待删除的数量，分页进行删除
        long count = dataMapper.selectCount(map);
        long num = count / DaoConst.SELECT_MAX_SIZE;
        long left = count % DaoConst.SELECT_MAX_SIZE;
        num = left > 0 ? num + 1 : num;
        long rows = DaoConst.SELECT_MAX_SIZE;
        List<Object> keyList = new ArrayList<>();
        for (long i = 0; i < num; i++) {
            long end = i == num - 1 ? count : (i + 1) * rows;
            map.put(DaoConst.FILTER, filter);
            keyList.addAll(tableBuilder.getKeyList(map, keyField, i * rows, rows, end));
        }
        if (CollectionUtils.isEmpty(keyList)) {
            return true;
        }
        return deleteByKeyList(keyList);
    }

    private void delEnableFalg(Map<String, Object> map) {
        map.remove(DaoConst.ENABLE_FLAG);
        map.remove(DaoConst.ENABLE_FLAG_VALUE);
        map.remove(DaoConst.ENABLE_FLAG_VALUE_NO);
    }
}
